package org.flink.api;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * @Author: TongRui乀
 * @Date: 2020/11/2 21:53
 * @description：  模拟实时生产数据流
 */
public class MyStreamSource implements SourceFunction<Item> {

    private boolean isRunning = true;

    /**
     *  每一秒生成一条数据
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Item> sourceContext) throws Exception {
        while (isRunning){
            Item item = generateItem();
            sourceContext.collect(item);
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    private Item generateItem(){
        int i = new Random().nextInt(100);
        Item item = new Item();
        item.setName("name" + i % 5);
        item.setId(i);
        item.setTimestamp(System.currentTimeMillis());
        return item;
    }

}
